package com.dongnao.network.nio.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 模拟Netty实现
 */
public class SimilarNettyServer {
    private ServerSocketChannel serverSocketChannel;
    //处理accept事件
    private ReactorThread[] mainReactorThreads = new ReactorThread[1];
    //处理读写事件
    private ReactorThread[] subReactorThreads = new ReactorThread[8];
    //处理业务的线程
    private static ExecutorService workPool = Executors.newCachedThreadPool();

    public static void main(String[] args) throws IOException {
        SimilarNettyServer server = new SimilarNettyServer();
        server.newGroup();
        server.initAndRegister();
        server.bind();
    }

    private void newGroup() throws IOException {
        //创建IO线程，处理客户端连接以后的socketChannel的IO读写
        for (int i = 0; i < subReactorThreads.length; i++) {
            subReactorThreads[i] = new ReactorThread() {
                @Override
                public void handle(SelectableChannel channel) throws IOException {
                    SocketChannel ch = (SocketChannel) channel;
                    ByteBuffer requestBuffer = ByteBuffer.allocate(1024);
                    while (ch.isOpen() && ch.read(requestBuffer) != -1) {
                        if (requestBuffer.position() > 0) break;
                    }
                    if (requestBuffer.position() == 0) {
                        //sk.cancel();
                        return;//没有数据了，则不继续后面的处理
                    }
                    requestBuffer.flip();
                    byte[] content = new byte[requestBuffer.limit()];
                    requestBuffer.get(content);
                    System.out.println(Thread.currentThread().getName() + " 收到数据，来自：" + ch.getRemoteAddress());

                    //业务操作
                    workPool.submit(() -> {});

                    //响应结果200
                    String response = "HTTP/1.1 200 OK\r\n" + "Content-Length: 11\r\n" + "Hello World";
                    ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
                    while (buffer.hasRemaining()) {
                        ch.write(buffer);
                    }
                }
            };
        }

        //只处理serverSocketChannel
        for (int i = 0; i < mainReactorThreads.length; i++) {
            mainReactorThreads[i] = new ReactorThread() {
                AtomicInteger incr = new AtomicInteger(0);

                @Override
                void handle(SelectableChannel channel) throws IOException {
                    ServerSocketChannel ch = (ServerSocketChannel) channel;
                    SocketChannel socketChannel = ch.accept();
                    socketChannel.configureBlocking(false);
                    //收到连接建立的通知后，分发给IO线程继续读取数据
                    int index = incr.getAndIncrement() % subReactorThreads.length;
                    ReactorThread workEventLoop = subReactorThreads[index];
                    workEventLoop.doStart();
                    SelectionKey selectionKey = workEventLoop.register(socketChannel);
                    selectionKey.interestOps(SelectionKey.OP_READ);
                    System.out.println(Thread.currentThread().getName() + " 收到新连接：" + socketChannel);
                }
            };
        }
    }

    private void initAndRegister() throws IOException {
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        int index = new Random().nextInt(mainReactorThreads.length);
        mainReactorThreads[index].doStart();
        SelectionKey selectionKey = mainReactorThreads[index].register(serverSocketChannel);
        selectionKey.interestOps(SelectionKey.OP_ACCEPT);
    }

    private void bind() throws IOException {
        serverSocketChannel.bind(new InetSocketAddress(8080));
        System.out.println("服务端启动，端口8080");
    }

    abstract class ReactorThread extends Thread {
        private Selector selector;
        volatile boolean running = false;

        public ReactorThread() throws IOException {
            selector = Selector.open();
        }

        //selector监听到事件后，调用这个方法
        abstract void handle(SelectableChannel channel) throws IOException;

        @Override
        public void run() {
            while (running) {
                try {
                    selector.select(5000);

                    //获取查询结果
                    Set<SelectionKey> selKeySet = selector.selectedKeys();
                    Iterator<SelectionKey> it = selKeySet.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        it.remove();
                        System.out.println(Thread.currentThread().getName() + "key isAcceptable: " + key.isAcceptable()
                                + ", isConnectable: " + key.isConnectable()
                                + ", isReadable: " + key.isReadable()
                                + ", isWritable: " + key.isWritable()
                                + ", isValid: " + key.isValid());

                        if ((key.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0)  {
                            try {
                                SelectableChannel channel = (SelectableChannel) key.attachment();
//                                channel.configureBlocking(false);
                                handle(channel);
                                if (!channel.isOpen()) {
                                    System.out.println("----客户端关闭了连接----");
                                    key.cancel();//取消这个key的订阅
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                                key.cancel();
                            }
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        private SelectionKey register(SelectableChannel channel) throws ClosedChannelException {
            SelectionKey selectionKey = channel.register(selector, 0, channel);
            return selectionKey;
        }

        private void doStart() {
            if (!running) {
                running = true;
                start();
            }
        }
    }
}
